package com.thiswhyme.stream.rabbit.receive;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.thiswhyme.stream.rabbit.constant.MqConstants;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.boot.autoconfigure.jms.JmsProperties;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;

/**
 * @description: 带重试的消息监听器
 * 注：消息内容类型：String，contentType：application/json，deliveryMode：持久化
 * @author: thiswhyme
 * @date: 2021/1/29
 */
@Slf4j
public abstract class AbstractRetryReceiver implements IReceiver {

    /**
     * 消息重发
     * @param message 收到的消息对象（必填）
     * @param retryTimes 重试次数（必填）
     * @param bizParams 业务参数，用于重试n次后仍然失败的自定义业务处理（非必填）
     * @param bizHeaders 自定义消息头（非必填）
     * @throws Exception
     */
    public void retry(Message<String> message, int retryTimes, Object bizParams, Map<String, Object> bizHeaders)
            throws Exception {
        Channel channel = (Channel) message.getHeaders().get(AmqpHeaders.CHANNEL);
        String exchangeName = (String) message.getHeaders().get(AmqpHeaders.RECEIVED_EXCHANGE);
        String routingKey = (String) message.getHeaders().get(AmqpHeaders.RECEIVED_ROUTING_KEY);
        int currentRetryTimes = (int) Optional.ofNullable(message.getHeaders().get(MqConstants.HEADER_RETRY_TIMES)).orElse(0);

        doRetry(channel, exchangeName, routingKey, message.getPayload() == null ? null : message.getPayload().getBytes(),
                currentRetryTimes, retryTimes, bizHeaders, bizParams);
    }

    /**
     * 消息重发
     * @param headers 收到的消息头（必填）
     * @param msgContent 待重发的消息内容（非必填）
     * @param retryTimes 重试次数（必填）
     * @param bizParams 业务参数，用于重试n次后仍然失败的自定义业务处理（非必填）
     * @param bizHeaders 自定义消息头（非必填）
     * @throws Exception
     */
    public void retry(MessageHeaders headers, String msgContent, int retryTimes, Object bizParams,
                      Map<String, Object> bizHeaders) throws Exception {
        Channel channel = (Channel) headers.get(AmqpHeaders.CHANNEL);
        String exchangeName = (String) headers.get(AmqpHeaders.RECEIVED_EXCHANGE);
        String routingKey = (String) headers.get(AmqpHeaders.RECEIVED_ROUTING_KEY);
        int currentRetryTimes = (int) Optional.ofNullable(headers.get(MqConstants.HEADER_RETRY_TIMES)).orElse(0);

        doRetry(channel, exchangeName, routingKey, msgContent == null ? null : msgContent.getBytes(), currentRetryTimes,
                retryTimes, bizHeaders, bizParams);
    }

    /**
     * 消息重发
     * @param channel 消息通道
     * @param exchangeName 交换机名
     * @param routingKey 路由键
     * @param msgContent 待重发的消息内容
     * @param currentRetryTimes 当前重试次数
     * @param retryTimes 总重试次数
     * @param bizHeaders 业务参数，用于重试n次后仍然失败的自定义业务处理
     * @param bizParams 自定义消息头
     * @throws IOException
     */
    private void doRetry(Channel channel, String exchangeName, String routingKey, byte[] msgContent,
                         int currentRetryTimes, int retryTimes, Map<String, Object> bizHeaders, Object bizParams)
            throws IOException {
        log.info("第【{}】次重试，exchangeName：{}，routingKey：{}", currentRetryTimes, exchangeName, routingKey);
        if (currentRetryTimes < retryTimes){
            bizHeaders = Optional.ofNullable(bizHeaders).orElse(new HashMap<>());
            bizHeaders.put(MqConstants.HEADER_RETRY_TIMES, currentRetryTimes + 1);
            AMQP.BasicProperties basicProperties = new AMQP.BasicProperties(MessageProperties.CONTENT_TYPE_JSON,
                    null, bizHeaders,
                    JmsProperties.DeliveryMode.PERSISTENT.getValue(), 0, null, null,
                    null, null, null, null, null, null, null);
            channel.basicPublish(exchangeName, routingKey, basicProperties, msgContent);
        } else {
            log.info("重试{}次后仍然失败，开始自定义处理逻辑，消息内容：{}", retryTimes, new String(msgContent));
            doIfRetryFail(bizParams);
        }
    }

    /**
     * 重试n次后仍然失败的自定义业务处理
     * @param bizParams
     */
    public abstract void doIfRetryFail(Object bizParams);
}
